package sc.analysis.aroon;



import java.io.IOException;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import sc.writable.PriceData;
import sc.writable.WritableArrayListFloat;



/**
 * Mapper of the Aroon analysis. Calculates Aroon Up and Aroon Down for the 
 * current period and the period before (to know if the two lines have crossed).
 * 
 * @author Quentin Rossé
 *
 */
public class AroonMapper extends Mapper<Text, PriceData, Text, AroonData>{
	
	/**
	 * Perdiods for the aroon analysis
	 */
	private static final int[] PERIODS = {25, 125};
	
	/**
	 * data that will be written to the reducer
	 */
	private static AroonData ad = new AroonData();
	
	
	public void map(Text key, PriceData value, Context context) throws IOException, InterruptedException{
		
		WritableArrayListFloat dayLows = value.getDayLowPrices();
		WritableArrayListFloat dayHighs = value.getDayHighPrices();
		
		//for a n day Aroon, we need n price data and n+1 if we want up/Down 
		//cross alerts (we need Aroon from the day before)
		System.out.println(key);
		int daysSinceHigh;
		int daysSinceLow;
		int prevDaysSinceHigh;
		int prevDaysSinceLow;
		
		for(int period : PERIODS){
			daysSinceHigh     = dayHighs.size()- highestValueIndice(dayHighs, period, dayHighs.size()-1) - 1;
			daysSinceLow      = dayLows.size() - lowestValueIndice(dayLows, period, dayLows.size()-1) - 1;
			prevDaysSinceHigh = dayHighs.size()- highestValueIndice(dayHighs, period, dayHighs.size()-2) - 2;
			prevDaysSinceLow  = dayLows.size() - lowestValueIndice(dayLows, period, dayLows.size()-2) - 2;
			System.out.println("dsl : " + daysSinceLow + "  dsh : " + daysSinceHigh);
			System.out.println("pdsl : " + prevDaysSinceLow + "  pdsh : " + prevDaysSinceHigh);
			
			if(prevDaysSinceLow >= 0 && prevDaysSinceHigh >= 0){
				ad.setPeriod(period);
				ad.setDown(100 * (period - daysSinceLow)/period);
				ad.setUp(100 * (period - daysSinceHigh)/period);
				ad.setPrevDown(100 * (period - prevDaysSinceLow)/period);
				ad.setPrevUp(100 * (period - prevDaysSinceHigh)/period);
				context.write(key, ad);
			}
		}
	}
	
	/**
	 * Searches the indice of the smallest value in the list
	 * 
	 * @param dayLows list of values
	 * @param period period of time for the calculation
	 * @param startIndice indice of the array where we start searching /!\ the
	 *                    first value is the highest indice
	 * @return
	 */
	private int lowestValueIndice(WritableArrayListFloat dayLows,
			                      int period,
			                      int startIndice){
		float lowestVal = Float.MAX_VALUE;
		int lowerstIndice = dayLows.size()-1;
		
		if(startIndice <= period || startIndice >= dayLows.size())
			return -1;
		else{
			for(int i=startIndice; i>=startIndice - period; i--){
				if(dayLows.getFloat(i) < lowestVal){
					lowestVal = dayLows.getFloat(i);
					lowerstIndice = i;
				}
			}
			return lowerstIndice;
		}
	}
	
	/**
	 * Searches the indice of the highest value in the list
	 * 
	 * @param dayHighs list of values
	 * @param period period of time for the calculation
	 * @param startIndice indice of the array where we start searching /!\ the
	 *                    first value is the highest indice
	 * @return
	 */
	private int highestValueIndice(WritableArrayListFloat dayHighs,
                                   int period,
 			                       int startIndice){
		float highestVal = 0f;
		int highestIndice = dayHighs.size()-1;
		
		if(startIndice <= period || startIndice >= dayHighs.size())
			return -1;
		else{
			for(int i=startIndice; i>=startIndice - period; i--){
				if(dayHighs.getFloat(i) > highestVal){
					highestVal = dayHighs.getFloat(i);
					highestIndice = i;
				}
			}
			return highestIndice;
		}
	}
	
	@Override
	public void setup(Context context){}
	
	@Override
	public void cleanup(Context context){}
	
}
